
// src/routes/command.js
const express = require("express");
const Queue = require("bull");
const cron = require("node-cron");
const Redis = require("ioredis");
const { v4: uuidv4 } = require("uuid");
const logger = require("../../utils/logger");
const platformSelector = require("../../utils/platformSelector");
// const auth = require("../../middleware/auth");
const errorHandler = require("../../middleware/errorHandler");
const notification = require("../../utils/notification");
const { saveData } = require("../../utils/index");

const dayjs = require("dayjs");
const chunk = require("lodash/chunk");
const axios = require("axios");

const router = express.Router();

// 创建Redis客户端
const redis = new Redis(process.env.REDIS_URL);
logger.info("Redis客户端已创建: " + process.env.REDIS_URL);

// 创建任务队列
const taskQueue = new Queue("taskQueue", { redis: process.env.REDIS_URL });
logger.info("任务队列已创建");

// 任务状态枚举
const TaskStatus = {
  PENDING: "pending",
  IN_PROGRESS: "in_progress",
  COMPLETED: "completed",
  FAILED: "failed",
  SCHEDULED: "scheduled",
};

// 存储定时任务的字典
const scheduledTasks = {};

// 更新任务状态
async function updateTaskStatus(taskId, status, message = "") {
  await redis.hset(`task:${taskId}`, "status", status, "message", message);
  logger.info(`任务 ${taskId} 状态已更新为 ${status}`);
}

// 添加过期时间
async function addExpiryToTask(taskId) {
  const taskInfo = await redis.hgetall(`task:${taskId}`);
  if (!taskInfo.schedule) {
    const expiryTimeInSeconds = 3 * 24 * 60 * 60; // 3 天过期
    await redis.expire(`task:${taskId}`, expiryTimeInSeconds);
    logger.info(`已为任务 ${taskId} 设置过期时间`);
  }
}

// 保存任务结果（修改为分批存储）
async function saveTaskResult(taskId, result) {
  const batchSize = 100000;
  const totalItems = result.length;
  const totalBatches = Math.ceil(totalItems / batchSize);

  logger.info(
    `正在保存任务 ${taskId} 的结果：${totalItems} 个项目分为 ${totalBatches} 批`
  );

  // 使用 multi 来确保原子性
  const multi = redis.multi();

  // 存储元数据
  multi.hmset(`task:${taskId}`, {
    totalItems,
    totalBatches,
    batchSize,
  });

  // 分批存储数据
  for (let i = 0; i < totalBatches; i++) {
    const start = i * batchSize;
    const end = Math.min((i + 1) * batchSize, totalItems);
    const batchData = result.slice(start, end);

    multi.del(`task:${taskId}:results`); // 先删除旧的数据
    multi.zadd(
      `task:${taskId}:results`,
      ...batchData.flatMap((item, index) => [
        start + index,
        JSON.stringify(item),
      ])
    );
    // 入库 mysql 
    logger.debug(`已保存第 ${i + 1} 批数据，共 ${end - start} 个项目`);
  }

  await multi.exec();
  logger.info(`任务 ${taskId} 的结果保存完成`);
  // try {
  //   logger.info(`开始持久化任务 ${taskId} 的结果`);
  //   // 保存任务结果
  //   await saveDataByBatch(taskId, result);
  // } catch (error) {
  //   logger.error(`持久库保存任务 ${taskId} 的结果时出错：${error}`);
  // }
}
// 分块持久存储
async function saveDataByBatch(taskId, result) {
  const batchSize = 100; // 每批100条数据
  const delay = 1000; // 每批之间延迟1秒
  const collected_at = dayjs().format("YYYY-MM-DD HH:mm:ss");

  const batches = chunk(result, batchSize);

  logger.info(`开始持久化任务 ${taskId} 的结果，共 ${batches.length} 批`);

  for (let i = 0; i < batches.length; i++) {
    try {
      await saveData({
        key: `${taskId}`,
        data: batches[i],
        collected_at,
      });
      logger.info(`成功保存任务 ${taskId} 的第 ${i + 1} 批数据`);

      if (i < batches.length - 1) {
        await new Promise((resolve) => setTimeout(resolve, delay));
      }
    } catch (error) {
      logger.error(
        `持久库保存任务 ${taskId} 的第 ${i + 1} 批数据时出错：${error}`
      );
    }
  }

  logger.info(`任务 ${taskId} 的所有数据批次已处理完毕`);
}

// 获取任务信息（不包含结果数据）
async function getTaskInfo(taskId) {
  const taskInfo = await redis.hgetall(`task:${taskId}`);
  delete taskInfo.result; // 移除结果数据，只返回元数据
  logger.info(`获取到任务 ${taskId} 的信息`);
  return taskInfo;
}

// 获取分页任务结果
async function getTaskResults(taskId, page = 1, pageSize = 100) {
  const taskInfo = await getTaskInfo(taskId);
  const totalItems = parseInt(taskInfo.totalItems || 0);

  if (totalItems === 0) {
    logger.info(`任务 ${taskId} 没有结果数据`);
    return { results: [], total: 0, page, pageSize };
  }

  const start = (page - 1) * pageSize;
  const end = Math.min(start + pageSize - 1, totalItems - 1);

  const results = await redis.zrange(`task:${taskId}:results`, start, end);
  logger.info(`获取任务 ${taskId} 的结果：第 ${page} 页，每页 ${pageSize} 项`);

  return {
    results: results.map(JSON.parse),
    total: totalItems,
    page,
    pageSize,
  };
}

// 睡眠函数
function sleep(s) {
  return new Promise((resolve) => setTimeout(resolve, s * 1000));
}

// 任务处理器 执行任务
taskQueue.process(async (job) => {
  const { taskId, platform, command, params } = job.data;

  return new Promise(async (resolve, reject) => {
    const timeoutId = setTimeout(() => {
      reject(new Error("任务执行超时"));
    }, 60 * 60 * 1000); // 60分钟超时

    try {
      logger.info(`开始处理任务 ${taskId}`);
      await updateTaskStatus(taskId, TaskStatus.IN_PROGRESS);

      // 选择并执行对应平台的处理逻辑
      const handler = platformSelector(platform);
      // console.log('<<<<<<<<<<<------------------>>>>>>>>', params)
      const data = await handler.handleCommand(command, params, taskId);

      // 保存任务结果
      await saveTaskResult(taskId, data.data);
      console.log('taskId:-->', taskId);
      console.log('data:--->', data.data.data);


      await updateTaskStatus(taskId, TaskStatus.COMPLETED);

      logger.info(`任务 ${taskId} 处理完成`);

      clearTimeout(timeoutId);
      resolve(data);
    } catch (error) {
      clearTimeout(timeoutId);
      logger.error(`处理任务 ${taskId} 时出错：${error}`);
      await updateTaskStatus(taskId, TaskStatus.FAILED, error.message);
      reject(error);
    }
  });
});

// 添加任务或定时任务
router.post("/command", async (req, res, next) => {
  try {
    const { platform, command, params, schedule } = req.body;
    const taskId = uuidv4(); // 生成唯一的taskId

    if (schedule) {
      // 创建定时任务
      const scheduledTask = cron.schedule(schedule, async () => {
        await taskQueue.add({ taskId, platform, command, params });
        logger.info(`执行定时任务，taskId: ${taskId}`);
      });

      scheduledTasks[taskId] = scheduledTask;

      // 将定时任务信息保存到Redis
      await redis.hmset(`task:${taskId}`, {
        status: TaskStatus.SCHEDULED,
        platform,
        command,
        params: JSON.stringify(params),
        schedule,
      });
      logger.info(`创建定时任务 ${taskId}`);
    } else {
      // 立即执行的任务
      await taskQueue.add({ taskId, platform, command, params });
      await updateTaskStatus(taskId, TaskStatus.PENDING);
      // 添加 platform 字段
      await redis.hset(`task:${taskId}`, "platform", platform);
      // 添加过期时间
      await addExpiryToTask(taskId);
      logger.info(`创建立即执行任务 ${taskId}`);
    }

    res.json({ status: "success", code: 200, taskId });
  } catch (error) {
    logger.error(`创建任务时出错：${error}`);
    next(error);
  }
});

// 获取任务信息和结果（分页）
async function getTaskInfoAndResults(taskId, page = 1, pageSize = 100) {
  const taskInfo = await redis.hgetall(`task:${taskId}`);

  if (!taskInfo.status) {
    logger.warn(`任务 ${taskId} 不存在`);
    return null; // 任务不存在
  }

  const totalItems = parseInt(taskInfo.totalItems || 0);
  let results = [];
  let total = 0;

  if (totalItems > 0) {
    const start = (page - 1) * pageSize;
    const end = Math.min(start + pageSize - 1, totalItems - 1);

    results = await redis.zrange(`task:${taskId}:results`, start, end);
    results = results.map(JSON.parse);
    total = totalItems;
    logger.info(`获取任务 ${taskId} 的结果：第 ${page} 页，共 ${total} 项`);
  }

  // 确保 params 字段是对象
  if (taskInfo.params) {
    try {
      taskInfo.params = JSON.parse(taskInfo.params);
    } catch (error) {
      logger.error(`解析任务 ${taskId} 的参数时出错：${error}`);
      taskInfo.params = {}; // 如果解析失败，设置为空对象
    }
  }

  delete taskInfo.result; // 移除可能存在的旧结果数据

  return {
    taskInfo,
    results: {
      data: results,
      total,
      page: parseInt(page),
      pageSize: parseInt(pageSize),
    },
  };
}

// 合并的获取任务状态和结果的接口
router.get("/task/:taskId", async (req, res, next) => {
  try {
    const { page = 1, pageSize = 10 } = req.query;
    const taskData = await getTaskInfoAndResults(
      req.params.taskId,
      parseInt(page),
      parseInt(pageSize)
    );

    if (taskData) {
      logger.info(`成功获取任务 ${req.params.taskId} 的信息和结果`);
      res.json({ status: "success", code: 200, ...taskData });
    } else {
      logger.warn(`未找到任务 ${req.params.taskId}`);
      res
        .status(404)
        .json({ status: "error", code: 404, message: "未找到任务" });
    }
  } catch (error) {
    logger.error(`获取任务信息时出错：${error}`);
    next(error);
  }
});
// 获取任务状态和元信息
// router.get("/task/:taskId", async (req, res, next) => {
//   try {
//     const taskInfo = await getTaskInfo(req.params.taskId);
//     if (taskInfo.status) {
//       res.json({ status: "success", code: 200, task: taskInfo });
//     } else {
//       res
//         .status(404)
//         .json({ status: "error", code: 404, message: "Task not found" });
//     }
//   } catch (error) {
//     next(error);
//   }
// });

// 获取任务结果（分页）
router.get("/task/:taskId/results", async (req, res, next) => {
  try {
    const { page = 1, pageSize = 10 } = req.query;
    const results = await getTaskResults(
      req.params.taskId,
      parseInt(page),
      parseInt(pageSize)
    );
    logger.info(
      `获取任务 ${req.params.taskId} 的结果：第 ${page} 页，每页 ${pageSize} 项`
    );
    res.json({ status: "success", code: 200, ...results });
  } catch (error) {
    logger.error(`获取任务结果时出错：${error}`);
    next(error);
  }
});

// 获取所有定时任务
router.get("/scheduled-tasks", async (req, res, next) => {
  try {
    const scheduledTaskKeys = await redis.keys("task:*");
    const scheduledTasksInfo = await Promise.all(
      scheduledTaskKeys.map(async (key) => {
        try {
          const keyType = await redis.type(key);
          if (keyType !== "hash") {
            return null;
          }

          const taskInfo = await redis.hgetall(key);
          if (taskInfo.schedule) {
            try {
              taskInfo.params = JSON.parse(taskInfo.params);
            } catch (parseError) {
              taskInfo.params = {}; // 设置为空对象以避免进一步的错误
            }
            return { id: key.split(":")[1], ...taskInfo };
          }
          return null;
        } catch (error) {
          logger.error(`处理任务 ${key} 时出错：${error.message}`);
          return null;
        }
      })
    );

    const filteredTasks = scheduledTasksInfo.filter((task) => task !== null);
    logger.info(`成功获取 ${filteredTasks.length} 个定时任务信息`);

    res.json({ status: "success", code: 200, scheduledTasks: filteredTasks });
  } catch (error) {
    logger.error(`获取定时任务列表时出错：${error.message}`);
    next(error);
  }
});

// 删除定时任务
router.delete("/task/:taskId", async (req, res, next) => {
  try {
    const { taskId } = req.params;
    const scheduledTask = scheduledTasks[taskId];
    if (scheduledTask) {
      scheduledTask.stop();
      delete scheduledTasks[taskId];
      await redis.del(`task:${taskId}`);
      logger.info(`定时任务 ${taskId} 已删除`);
      res.json({ status: "success", code: 200, message: "定时任务已删除" });
    } else {
      logger.warn(`未找到定时任务 ${taskId}`);
      res
        .status(404)
        .json({ status: "error", code: 404, message: "没找到定时任务" });
    }
  } catch (error) {
    logger.error(`删除定时任务时出错：${error}`);
    next(error);
  }
});
// 触发指定的定时任务
router.post("/trigger-scheduled-task/:taskId", async (req, res, next) => {
  try {
    const { taskId } = req.params;
    const taskInfo = await redis.hgetall(`task:${taskId}`);

    if (!taskInfo.schedule) {
      logger.warn(`任务 ${taskId} 不是定时任务`);
      return res.status(400).json({
        status: "error",
        code: 400,
        message: "指定的任务不是定时任务",
      });
    }

    // 立即执行定时任务
    await taskQueue.add({
      taskId,
      platform: taskInfo.platform,
      command: taskInfo.command,
      params: JSON.parse(taskInfo.params),
    });

    logger.info(`手动触发定时任务 ${taskId} 执行`);
    res.json({
      status: "success",
      code: 200,
      message: "定时任务已触发执行",
      platform: taskInfo.platform,
    });
  } catch (error) {
    logger.error(`触发定时任务时出错：${error}`);
    next(error);
  }
});

// 启动时恢复所有定时任务
async function restoreScheduledTasks() {
  const taskKeys = await redis.keys("task:*");
  for (const key of taskKeys) {
    if ((await redis.type(key)) !== "hash") continue; // 跳过非哈希类型的键
    const taskInfo = await redis.hgetall(key);
    if (taskInfo.schedule) {
      const taskId = key.split(":")[1];
      const scheduledTask = cron.schedule(taskInfo.schedule, async () => {
        await taskQueue.add({
          taskId,
          platform: taskInfo.platform,
          command: taskInfo.command,
          params: JSON.parse(taskInfo.params),
        });
        logger.info(`执行定时任务，taskId: ${taskId}`);
      });
      scheduledTasks[taskId] = scheduledTask;
      logger.info(`恢复定时任务，taskId: ${taskId}`);
    }
  }
}

restoreScheduledTasks().catch((err) =>
  logger.error(`恢复定时任务时出错：${err.message}`)
);

//使用node-cron 定时任务，每五分钟检查一次任务队列状态
cron.schedule("*/5 * * * *", async () => {
  const jobCounts = await taskQueue.getJobCounts();
  logger.info(`当前任务队列状态：待处理任务${jobCounts.waiting}个`);
  if (jobCounts.waiting > 20) {
    await notification(
      "任务队列积压",
      `${process.env.NODE_ENV}环境下的任务队列积压`,
      "请及时查看任务处理情况"
    );
  }
});

router.use(errorHandler);

module.exports = router;
